package cn.lazar.homework

import java.util.Properties

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}

import scala.collection.mutable.ArrayBuffer

object SparkStream {
  def main(args: Array[String]): Unit = {
    // 初始化
    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf = new SparkConf().setAppName("FileDStream").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(10))


    // 定义kafka相关参数
//    val groupId: String = "group1"
//    val topics: Array[String] = Array("sample_input")
//    val kafkaParams: Map[String, Object] = getKafkaConsumerParameters(groupId)
//

    // 创建DStream
    val dstream: DStream[String] = ssc.socketTextStream("localhost", 9999)
//    // 创建DStream
//    val dstream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
//      ssc,
//      LocationStrategies.PreferConsistent,
//      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
//    )

//    val producer = getProductor()
    // DStream转换&输出

//   dstream.foreachRDD{ (rdd, time) =>
//        println(rdd.collect().toBuffer)
//      rdd.map(line=>{
//          val fields = line.split("<<<!>>>,<<<!>>>")
//          val arr = new ArrayBuffer[String]()
//          fields.foreach(item=>{
//             if (item.contains("<<<!>>>")){
//                 val s = item.replace("<<<!>>>","")
//                 arr.append(s)
//             }else if (item.length > 0){
//                 arr.append(item)
//             }else{
//
//             }
//          })
//          var str = ""
//          for (elem <- arr){
//             str = str + "|" +  elem
//          }
//          //sendMsg(producer,"sample_output",str)
//          println(str)
//      })
//       println(s"*********** rdd.count = ${rdd.count()}; time = $time *************")
//    }
val value = dstream.map(line => {
  val fields = line.split("<<<!>>>,<<<!>>>")
  val arr = new ArrayBuffer[String]()
  fields.foreach(item=>{
     if (item.contains("<<<!>>>")){
         val s = item.replace("<<<!>>>","")
         arr.append(s)
     }else if (item.length > 0){
         arr.append(item)
     }else{

     }
  })
  var str = ""
  for (elem <- arr){
     if (str.equals("")){
       str = elem
     }else{
       str = str + "|" +  elem
     }

  }
  //sendMsg(producer,"sample_output",str)
  println(str)
  str
})

    value.print()
    value.saveAsTextFiles("data/output")

    // 启动作业
    ssc.start()
    ssc.awaitTermination()
  }

  def getKafkaConsumerParameters(groupid: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node51:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupid,
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean),
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest"
    )
  }
  // 创建kafka生产者
  def getProductor(): KafkaProducer[String,String] ={
    // 定义 kafka 参数
    val brokers = "node51:9092"
    val prop = new Properties()

    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    // KafkaProducer
    val producer = new KafkaProducer[String, String](prop)

    producer
  }
  // 发送消息
  def sendMsg(producer: KafkaProducer[String,String],topic:String,msg: String): Unit ={
//    val msg = new ProducerRecord[String, String](topic, "",msg)
//    // 发送消息
//    producer.send(msg)
  }
}
